/*
 * Copyright 2011 LMAX Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.lmax.disruptor;


/**
 * {@link SequenceBarrier} handed out for gating {@link EventProcessor}s on a cursor sequence and optional dependent {@link EventProcessor}(s),
 * using the given WaitStrategy.
 */
final class ProcessingSequenceBarrier implements SequenceBarrier
{
    private final WaitStrategy waitStrategy;
    private final Sequence dependentSequence; // 可能是单个 Sequence（生产者光标） 也可能是一组Sequence（ FixedSequenceGroup ）
    private volatile boolean alerted = false;
    private final Sequence cursorSequence;
    private final Sequencer sequencer;

    ProcessingSequenceBarrier(
        final Sequencer sequencer, // 生产者Sequencer
        final WaitStrategy waitStrategy,// 消费者等待策略
        final Sequence cursorSequence, // 生产者的光标
        final Sequence[] dependentSequences)// 默认 new Sequences[0]
    {
        // 只在AbstractSequencer.newBarrier中被创建
        // public SequenceBarrier newBarrier(final Sequence... sequencesToTrack)
        //    {
        //        return new ProcessingSequenceBarrier(this, waitStrategy, cursor, sequencesToTrack);
        //    }
        this.sequencer = sequencer;
        this.waitStrategy = waitStrategy;
        this.cursorSequence = cursorSequence;

        if (0 == dependentSequences.length)
        {
            // 该ProcessingSequenceBarrier没有依赖的SequenceBarrier
            // 是 生产者Sequencer 的串行链路的第一批消费者节点
            // 所以直接指向生产者的游标
            dependentSequence = cursorSequence;
        }
        else
        {
            // 该ProcessingSequenceBarrier有依赖的SequenceBarrier（有前继节点）
            // 是 生产者Sequencer 的串行链路的第一批消费者节点
            // 所以直接指向前继节点的消费者组的游标的最小值（用于每次获取）
            dependentSequence = new FixedSequenceGroup(dependentSequences);
            // FixedSequenceGroups实现了Sequence接口 但是只支持get方法
            // public long get(){return Util.getMinimumSequence(sequences);}
        }
    }

    @Override
    public long waitFor(final long sequence) // 由EventProcessor（消费者）调用
        throws AlertException, InterruptedException, TimeoutException
    {

        checkAlert();// 只能在alerted==false时方法执行

        // Wait for the given sequence to be available。
        // dependentSequence ---> 该消费组节点的前继节点
        // （第一个消费者组前继节点是生产者Sequencer的cursor 后面的就是前继节点消费进度的的最小Sequence）
        // sequence参数是该消费者期望获取的下一个序号值 ---> 等待该序号可供消费
        // 这里并不保证返回值availableSequence一定等于 given sequence，他们的大小关系取决于采用的WaitStrategy。
        long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);

        //如果当前可用的序号小于期望获取的下一个序号，则返回availableSequence，
        // 这将导致调用者EventProcessor继续wait
        if (availableSequence < sequence)
        {
            return availableSequence;
        }

        // 获取消费者可以消费的最大的可用序号，支持批处理效应，提升处理效率。
        // 当availableSequence > sequence时，需要遍历 sequence --> availableSequence，找到最前一个准备就绪，可以被消费的event对应的seq。
        // 最小值为：sequence-1
        return sequencer.getHighestPublishedSequence(sequence, availableSequence);
    }

    @Override
    public long getCursor()
    {
        return dependentSequence.get();
    }

    @Override
    public boolean isAlerted()
    {
        return alerted;
    }

    @Override
    public void alert()
    {
        alerted = true;
        waitStrategy.signalAllWhenBlocking();
    }

    @Override
    public void clearAlert()
    {
        alerted = false;
    }

    @Override
    public void checkAlert() throws AlertException
    {
        if (alerted)
        {
            throw AlertException.INSTANCE;
        }
    }
}